package rabitmq

import (
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/lv_baobao/gcore/logHelper"
	"github.com/streadway/amqp"
	"strconv"
	"sync"
	"time"
)

type rabbitMqProducer struct {
	isConnected bool
	conn        *amqp.Connection
	//channel     *amqp.Channel
	lock sync.Mutex
}

func NewRabbitMqProducer() *rabbitMqProducer {
	r := new(rabbitMqProducer)
	r.lock = sync.Mutex{}
	return r
}

func (bus *rabbitMqProducer) tryConnect() (bool, error) {
	if bus.conn == nil || bus.conn.IsClosed() {
		bus.lock.Lock()
		defer bus.lock.Unlock()
		if bus.conn == nil || bus.conn.IsClosed() {
			logHelper.Info("开始尝试连接mq")
			connectUrl := fmt.Sprintf("amqp://%s:%s@%s:%d/%s", rabbitMqConfig.UserName, rabbitMqConfig.PassWord, rabbitMqConfig.Host, rabbitMqConfig.Port, rabbitMqConfig.VirtualHost)
			conn, err := amqp.Dial(connectUrl)
			if err != nil {
				logHelper.Error(err)
				return false, err
			}
			bus.conn = conn
			bus.isConnected = true
			logHelper.Info("连接mq成功", connectUrl)
		}
	}

	return true, nil
}

//func (bus *rabbitMqProducer) initDirect2(exchangeOpt ExchangeOption, queueOpt QueueOption) error {
//	if bus.conn == nil || bus.conn.IsClosed() {
//		bus.tryConnect()
//	}
//	if bus.conn == nil {
//		logHelper.Error("rabbitMq连接失败")
//		return errors.New("rabbitMq连接失败")
//	}
//	var err error
//	//defer bus.conn.Close()
//	if bus.channel == nil {
//		bus.lock.Lock()
//		defer bus.lock.Unlock()
//		bus.channel, err = bus.conn.Channel()
//		if err != nil {
//			logHelper.Error(err)
//			return err
//		}
//	}
//
//	bus.channel.ExchangeDeclare(
//		exchangeOpt.Name,
//		amqp.ExchangeDirect,
//		exchangeOpt.Durable,
//		exchangeOpt.AutoDelete,
//		false,
//		true,
//		exchangeOpt.Arguments,
//	)
//
//	_, err = bus.channel.QueueDeclare(
//		queueOpt.Name,       // name
//		queueOpt.Durable,    // durable
//		queueOpt.AutoDelete, // delete when unused
//		queueOpt.Exclusive,  // exclusive
//		false,               // no-wait
//		queueOpt.Arguments,  // arguments
//	)
//	if err != nil {
//		logHelper.Error(err)
//		return err
//	}
//	if queueOpt.RoutingKey == "" {
//		queueOpt.RoutingKey = queueOpt.Name
//	}
//	err = bus.channel.QueueBind(
//		queueOpt.Name,
//		queueOpt.RoutingKey,
//		exchangeOpt.Name,
//		false,
//		nil,
//	)
//	if err != nil {
//		logHelper.Error(err)
//		return err
//	}
//	return nil
//}

func (bus *rabbitMqProducer) initDirect(exchangeOpt ExchangeOption, queueOpt QueueOption) (*amqp.Channel, error) {
	var err error
	bus.conn, err = NewRabbitMqConnPool().GetFreeRabbitMqConn(time.Second * 2)
	if err != nil {
		logHelper.Error(err)
		return nil, errors.New("InitDirect error")
	}
	if bus.conn == nil {
		return nil, errors.New("InitDirect error ,conn is nil")
	}
	defer func() {
		if err != nil {
			//回收conn到池子
			NewRabbitMqConnPool().Free(bus.conn)
		}
	}()
	var channel *amqp.Channel
	channel, err = bus.conn.Channel()
	if err != nil {
		logHelper.Error(err)
		return nil, errors.New("InitDirect error")
	}
	err = channel.ExchangeDeclare(
		exchangeOpt.Name,
		amqp.ExchangeDirect,
		exchangeOpt.Durable,
		exchangeOpt.AutoDelete,
		false,
		true,
		exchangeOpt.Arguments,
	)
	if err != nil {
		logHelper.Error(err)
		return nil, err
	}
	_, err = channel.QueueDeclare(
		queueOpt.Name,       // name
		queueOpt.Durable,    // durable
		queueOpt.AutoDelete, // delete when unused
		queueOpt.Exclusive,  // exclusive
		false,               // no-wait
		queueOpt.Arguments,  // arguments
	)
	if err != nil {
		logHelper.Error(err)
		return nil, err
	}
	if queueOpt.RoutingKey == "" {
		queueOpt.RoutingKey = queueOpt.Name
	}
	err = channel.QueueBind(
		queueOpt.Name,
		queueOpt.RoutingKey,
		exchangeOpt.Name,
		false,
		nil,
	)
	if err != nil {
		logHelper.Error(err)
		return nil, err
	}
	return channel, nil
}

func (bus *rabbitMqProducer) PublishDirect(obj interface{}, exchangeOpt ExchangeOption, queueOpt QueueOption, expire int) error {
	s := strconv.Itoa(expire)
	var err error
	var channel *amqp.Channel
	channel, err = bus.initDirect(exchangeOpt, queueOpt)
	if err != nil {
		return err
	}
	defer NewRabbitMqConnPool().Free(bus.conn)
	defer channel.Close()
	var body, _ = json.Marshal(obj)
	var publish = amqp.Publishing{
		ContentType:  "text/plain",
		DeliveryMode: 2,
		Body:         body,
	}
	if expire > 0 {
		publish.Expiration = s
	}
	// mandatory true 未找到队列返回给消费者
	returnChan := make(chan amqp.Return, 0)
	channel.NotifyReturn(returnChan)

	err = channel.Confirm(false)
	if err != nil {
		return err
	}
	err = channel.Publish(
		exchangeOpt.Name, // exchange
		queueOpt.Name,    // routing key
		true,             // mandatory
		false,            // immediate
		publish)
	return err
}

func (bus *rabbitMqProducer) PublishFanout(obj interface{}, exchangeOpt ExchangeOption) error {
	var err error
	bus.conn, err = NewRabbitMqConnPool().GetFreeRabbitMqConn(time.Second * 2)
	if err != nil {
		logHelper.Error(err)
		return errors.New("GetFreeRabbitMqConn error")
	}
	if bus.conn == nil {
		return errors.New("GetFreeRabbitMqConn error")
	}
	defer NewRabbitMqConnPool().Free(bus.conn)
	var channel *amqp.Channel
	channel, err = bus.conn.Channel()
	if err != nil {
		logHelper.Error(err)
		return errors.New("create channel error")
	}
	defer func() {
		channel.Close()
		channel = nil
	}()
	err = channel.ExchangeDeclare(
		exchangeOpt.Name,
		amqp.ExchangeFanout,
		exchangeOpt.Durable,
		exchangeOpt.AutoDelete,
		false,
		true,
		exchangeOpt.Arguments,
	)
	if err != nil {
		logHelper.Error(err)
		return err
	}
	var body, _ = json.Marshal(obj)
	err = channel.Publish(
		exchangeOpt.Name, // exchange
		"",               // routing key
		false,            // mandatory
		false,            // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        body,
		})
	return err
}
